package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.courseOrderCount;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.DimAsyncFunction;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @Author: LixuanFeng
 * @Date: 2022/9/2 13:53
 * @From: GoodGoodStudyDayDayUp
 *
 *  复习： kafka upsert表里面有之前回撤流产生的多余数据，left join 之后保留left表所有数据 join 上 右表就会产生重复的数据
 *
 *    TODO 去重：
 *
 *       下单详情表的来源: 预处理表
 *         做了个过滤
 *
 *     预处理表是如何实现?
 *             order_detail
 *                 join
 *             order_info
 *                left join
 *             详情活动
 *                 left join
 *             详情优惠券
 *                 look up join
 *             字典表
 *     因为有left join, 对消费者来说, 同一个详情会可能产生重复数据
 *
 *     去重思路：
 *             2. 定时器
 *             同一个详情id, 第一条数据进来的时候注册一个 5s 后触发的定时器
 *             每来一条数据, 把数据存入到状态中, 后面来的数据和状态中的做比较, 时间变大, 则更新状态
 *
 *             等到定时器触发的时候, 则最大的比较出来了
 *
 *             会话窗口和定时器哪个更优?
 *
 *                定时器更优: 时效性更好
 *
 *         3. 如果在统计聚合计算的时候, 根本就用不到右表的数据, 就不需要等最全的了
 *             用第一条数据
 *
 *      TODO 这次没有用到left join，原则上使用第一条数据就可以， 但是为了逻辑完整，还是要进行去重操作
 *
 *      TODO doris 可以 rollup 和 物化视图，所以这里只需要按照最小粒度课程聚合，
 *       然后补充维度后 科目 类型 ，再以这些维度，创建聚合表即可
 *
 *      doris建表语句
 *              drop table if exists dws_course_order_count_window;
 *              create table if not exists dws_course_order_count_window
 *              (
 *                  `stt`            DATETIME comment '窗口起始时间',
 *                  `edt`            DATETIME comment '窗口结束时间',
 *                  `cur_date`       DATE comment '当天日期',
 *                  `course_id`       VARCHAR(128) comment'课程id',
 *                  `course_name`      VARCHAR(128)  comment'课程name',
 *                  `subject_id`       VARCHAR(128)  comment'科目id',
 *                  `subject_name`      VARCHAR(128)  comment'科目Name',
 *                  `category_id`       VARCHAR(128)  comment'分类id',
 *                  `category_name`     VARCHAR(128)   comment'分类name',
 *                  `order_ct`          BIGINT replace comment '下单次数',
 *                  `order_unique_ct`    BIGINT replace comment '下单人数',
 *                  `order_amount`       DECIMAL(16, 2) replace comment '订单金额'
 *              ) engine = olap aggregate key (`stt`, `edt`, `cur_date`,`courseId`,`courseName`,`subjectId`,
 *              `subjectName`,`categoryId`,`categoryName`)
 *              comment "课程主题下单汇总表"
 *              partition by range(`cur_date`)()
 *              distributed by hash(`stt`) buckets 10 properties (
 *                "replication_num" = "3",
 *                "dynamic_partition.enable" = "true",
 *                "dynamic_partition.time_unit" = "DAY",
 *                "dynamic_partition.start" = "-1",
 *                "dynamic_partition.end" = "1",
 *                "dynamic_partition.prefix" = "par",
 *                "dynamic_partition.buckets" = "10",
 *                "dynamic_partition.hot_partition_num" = "1"
 *              );
 */
public class DwsCourseOrderCountWindow_lixuanfeng extends BaseAppOneStream {

    public static void main(String[] args) {

        new DwsCourseOrderCountWindow_lixuanfeng().init(3111,2,
                "DwsCourseOrderCountWindow_lixuanfeng"
        , Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);

    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.数据去重
        SingleOutputStreamOperator<JSONObject> distStream = distinctData(stream);
    //    distStream.print("!!!!!!!!!!!!!!!!!");

        //2.按uid分组统计，封装到pojo
        SingleOutputStreamOperator<courseOrderCount> countByUserPojo = countAndParseToPojo(distStream);
     //  countByUserPojo.print("!!!!!!!");

        //3.再根据courseid分组统计，开窗聚合
       SingleOutputStreamOperator<courseOrderCount> pojoAggStream = countByCourseId(countByUserPojo);
       //TODO 测试OK

        //4.补充维度
        SingleOutputStreamOperator<courseOrderCount> afterAddDimStream = addDim(pojoAggStream);
       //afterAddDimStream.print("oooooooooo");
        //TODO 补充改维度测试OK

        //5.写入doris
       writeToDoris(afterAddDimStream);


    }

    private void writeToDoris(SingleOutputStreamOperator<courseOrderCount> afterAddDimStream) {
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        String tableName= "edu.dws_course_order_count_window";

        afterAddDimStream
                .map(pojo -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(pojo, config);
                })
                .addSink(DorisSink.sink(new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .setTableIdentifier(tableName)
                                .build()));
    }

    private SingleOutputStreamOperator<courseOrderCount> addDim(SingleOutputStreamOperator<courseOrderCount> pojoAggStream) {

        //添加维度   redis缓存  + 异步
        //从DIM_COURSE_INFO先添加到 学科id
        SingleOutputStreamOperator<courseOrderCount> subjectIdStream = AsyncDataStream.unorderedWait(
                pojoAggStream,
                new DimAsyncFunction<courseOrderCount>() {
                    @Override
                    public String getTable() {
                        return "DIM_COURSE_INFO";
                    }

                    @Override
                    public String getId(courseOrderCount input) {
                        return input.getCourseId();
                    }

                    @Override
                    public void addDim(JSONObject dim, courseOrderCount input) {
                        input.setSubjectId(dim.getString("SUBJECT_ID"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<courseOrderCount> categroyIdStream = AsyncDataStream.unorderedWait(
                subjectIdStream,
                new DimAsyncFunction<courseOrderCount>() {
                    @Override
                    public String getTable() {
                        return "DIM_BASE_SUBJECT_INFO";
                    }

                    @Override
                    public String getId(courseOrderCount input) {
                        return input.getSubjectId();
                    }

                    @Override
                    public void addDim(JSONObject dim, courseOrderCount input) {
                        input.setSubjectName(dim.getString("SUBJECT_NAME"));
                        input.setCategoryId(dim.getString("CATEGORY_ID"));
                    }
                }
                , 60L,
                TimeUnit.SECONDS
        );

      return  AsyncDataStream.unorderedWait(
                categroyIdStream,
                new DimAsyncFunction<courseOrderCount>() {
                    @Override
                    public String getTable() {
                        return "DIM_BASE_CATEGORY_INFO";
                    }

                    @Override
                    public String getId(courseOrderCount input) {
                        return input.getCategoryId();
                    }

                    @Override
                    public void addDim(JSONObject dim, courseOrderCount input) {
                        input.setCategoryName(dim.getString("CATEGORY_NAME"));
                    }
                }
                , 60L,
                TimeUnit.SECONDS
        );

    }

    private SingleOutputStreamOperator<courseOrderCount> countByCourseId(SingleOutputStreamOperator<courseOrderCount> countByUserPojo) {
          return
            countByUserPojo
                    .keyBy(pojo->pojo.getCourseId())
                    .window(TumblingEventTimeWindows.of(Time.seconds(5L)))
                    .reduce(new ReduceFunction<courseOrderCount>() {
                                @Override
                                public courseOrderCount reduce(courseOrderCount value1, courseOrderCount value2) throws Exception {
                                  //  System.out.println("OKOKOKOKOK");
                                    value1.setOrderCt(value1.getOrderCt() + value2.getOrderCt());
                                    value1.setOrderUniqueCt(value1.getOrderUniqueCt() + value2.getOrderUniqueCt());
                                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));

                                    return value1;
                                }
                            },
                            new ProcessWindowFunction<courseOrderCount, courseOrderCount, String, TimeWindow>() {
                                @Override
                                public void process(String s, Context context, Iterable<courseOrderCount> elements, Collector<courseOrderCount> out) throws Exception {
                                    courseOrderCount pojo = elements.iterator().next();


                                    String start = AtguiguUtil.toDate(context.window().getStart());
                                    String end = AtguiguUtil.toDate(context.window().getEnd());
                                    String curDate = AtguiguUtil.toDate(System.currentTimeMillis());

                                    pojo.setStt(start);
                                    pojo.setEdt(end);
                                    pojo.setCurDate(curDate);

                                    out.collect(pojo);

                                  //  System.out.println("xxxxxxxxxxxxxxxxxxxxxxx"+pojo);
                                }
                            }
                    );
    }



    private SingleOutputStreamOperator<courseOrderCount> countAndParseToPojo(SingleOutputStreamOperator<JSONObject> distStream) {
        return
        distStream
                .assignTimestampsAndWatermarks( WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((json,ts)->json.getLong("od_ts")*1000))
                //先按照user_id 求出去重的人数
                .keyBy(json->json.getString("user_id"))

                .process(
                        new KeyedProcessFunction<String, JSONObject, courseOrderCount>() {

                    private ValueState<String> userLastOdDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        userLastOdDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("userLastOdDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<courseOrderCount> out) throws Exception {
                        String lastOdDate = userLastOdDate.value();
                        Long nowOdTs = value.getLong("od_ts");
                        String nowOdDate = AtguiguUtil.toDate(nowOdTs);


                        Long orderCt=1L;
                        Long orderUniqueCt=0L;
                        BigDecimal orderAmount = value.getBigDecimal("split_final_amount");

                        if (lastOdDate==null || !nowOdDate.equals(lastOdDate)){
                            orderUniqueCt=1L;
                            userLastOdDate.update(nowOdDate);
                        }

                        out.collect(courseOrderCount.builder()
                                .courseId(value.getString("course_id"))
                                .courseName(value.getString("course_name"))
                                .orderCt(orderCt)
                                .orderAmount(orderAmount)
                                .orderUniqueCt(orderUniqueCt)
                                .build());
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> distinctData(DataStreamSource<String> stream) {
     return
        stream
                .map(jsonStr-> JSON.parseObject(jsonStr));
//                .keyBy(json->json.getString("id"))
//                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
//
//                    private JSONObject maxTsData;
//                    private ValueState<JSONObject> maxTsDateState;
//
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        maxTsDateState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("maxTsDateState", JSONObject.class));
//                    }
//
//                    @Override
//                    public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
//                        //状态维护的是一条数据
//                        maxTsData = maxTsDateState.value();
//
//                        // 定时器触发的时候, 把状态中的数据输出
//                        if (maxTsData ==null){
//                            // 第一条数据来的时候, 把数据存到状态, 注册5s后触发的定时器
//                            maxTsDateState.update(value);
//                            ctx.timerService().registerEventTimeTimer(ctx.timerService().currentProcessingTime()+5000);
//                        } else {
//                            //如果不是第一条，与状态中的数据做对比
//                            String lastRowOpTs = maxTsData.getString("row_op_ts");
//                            String nowRowOpTs = value.getString("row_op_ts");
//
//                            if (AtguiguUtil.isLarger(nowRowOpTs,lastRowOpTs)){
//                            /// 判断current时间是否大于last, 返回true表示大
//                                maxTsDateState.update(value);
//                            }
//                        }
//                    }
//
//                    @Override
//                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
//                        out.collect(maxTsData);
//                    }
//                });

    }
}
